package mrunit;

import com.ailk.oci.ocnosql.common.rowkeygenerator.GenRKCallBackHBImpl;
import com.ailk.oci.ocnosql.common.rowkeygenerator.MD5RowKeyGenerator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * Created by zhaopan on 2017/6/27.<br>
 *
 * @Description:
 */
public class IndexCoprocessor4 extends BaseRegionObserver
{
    @Override
    public void prePut(final ObserverContext<RegionCoprocessorEnvironment> e,
                       final Put put, final WALEdit edit, final Durability durability)
            throws IOException
    {
        Configuration conf = new Configuration();

        String tableName = e.getEnvironment().getRegionInfo().getTable().toString();
        //ConnectionFactory.createConnection(conf).getTable(TableName.valueOf(tableName));
        System.out.println("put.size===="+put.size());
        HTable table = new HTable(conf, tableName + "_INDEX");
        List<Cell> imeis = put.get("0".getBytes(), "CALLINGIMEI".getBytes());
        List<Cell> calleds = put.get("0".getBytes(), "CALLEDNUM".getBytes());
        List<Cell> btimes = put.get("0".getBytes(), "BTIME".getBytes());
        //System.out.println("imeis.size===="+imeis.size());
        ///System.out.println("calleds.size===="+calleds.size());
        ///System.out.println("btimes.size===="+btimes.size());
        int lenth=Math.min(Math.min(imeis.size(),calleds.size()),btimes.size());
        //System.out.println("lenth===="+lenth);
        for(int i=0;i<lenth;i++){
            String btime = Bytes.toString(CellUtil.cloneValue(btimes.get(i)));
            String imei = Bytes.toString(CellUtil.cloneValue(imeis.get(i)));
            if (imei != null && !"".equals(imei))
            {
                String imeiMd5 = new MD5RowKeyGenerator().generatePrefix(imei).toString();
                String imeiRowKey = new GenRKCallBackHBImpl().callback(imeiMd5 + "!" + imei + "" + btime, imei + "" + btime, null);
                Put imeiIndexPut = new Put(imeiRowKey.getBytes());
                imeiIndexPut.addColumn("0".getBytes(), "IX".getBytes(), CellUtil.cloneRow(imeis.get(i)));
                table.put(imeiIndexPut);
            }
            String called = Bytes.toString(CellUtil.cloneValue(calleds.get(i)));
            if (called != null && !"".equals(called))
            {
                String calledMd5 = new MD5RowKeyGenerator().generatePrefix(called).toString();
                String calledRowKey = new GenRKCallBackHBImpl().callback(calledMd5 + "@" + called + "" + btime, called + "" + btime, null);
                Put calledIndexPut = new Put(calledRowKey.getBytes());
                calledIndexPut.addColumn("0".getBytes(), "IX".getBytes(), CellUtil.cloneRow(calleds.get(i)));
                table.put(calledIndexPut);
            }
        }
        table.close();
        /*
        for (int i = 0; i < imeis.size(); i++)
        {
            String btime = Bytes.toString(CellUtil.cloneValue(btimes.get(i)));
            String imei = Bytes.toString(CellUtil.cloneValue(imeis.get(i)));
            if (imei != null && !"".equals(imei))
            {

                String imeiMd5 = new MD5RowKeyGenerator().generatePrefix(imei).toString();
                String imeiRowKey = new GenRKCallBackHBImpl().callback(imeiMd5 + "" + imei + "" + btime, imei + "" + btime, null);
                Put imeiIndexPut = new Put(imeiRowKey.getBytes());
                imeiIndexPut.addColumn("0".getBytes(), "IX".getBytes(), CellUtil.cloneRow(imeis.get(i)));
                table.put(imeiIndexPut);
            }
            String called = Bytes.toString(CellUtil.cloneValue(calleds.get(i)));
            if (called != null && !"".equals(called))
            {
                String calledMd5 = new MD5RowKeyGenerator().generatePrefix(called).toString();
                String calledRowKey = new GenRKCallBackHBImpl().callback(calledMd5 + "" + called + "" + btime, called + "" + btime, null);
                Put calledIndexPut = new Put(calledRowKey.getBytes());
                calledIndexPut.addColumn("0".getBytes(), "IX".getBytes(), CellUtil.cloneRow(calleds.get(i)));
                table.put(calledIndexPut);
            }


        }
        */
    }

}
